package com.shaonaiyi.hadoop.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.net.URI;

/**
 * @Auther: 邵奈一
 * @Date: 2019/03/21 下午 4:30
 * @Description: Java API实现HDFS的相关操作
 */
public class HDFSClient {

    public static final String HDFS_PATH = "hdfs://master:9999";

    FileSystem fileSystem = null;
    Configuration configuration = null;

    /**
     * 创建HDFS文件夹
     * @throws Exception
     */
    @Test
    public void mkdir() throws Exception {
        fileSystem.mkdirs(new Path("/JavaAPI"));
    }

    /**
     * 创建HDFS文件
     * @throws Exception
     */
    @Test
    public void create() throws Exception {
        FSDataOutputStream output = fileSystem.create(new Path("/JavaAPI/JavaAPI.txt"));
        output.write("hello,shaonaiyi".getBytes());
        output.flush();
        output.close();
    }

    /**
     * 查看HDFS上的文件内容
     * @throws Exception
     */
    @Test
    public void cat() throws Exception {
        FSDataInputStream in = fileSystem.open(new Path("/JavaAPI/JavaAPI.txt"));
        IOUtils.copyBytes(in, System.out, 1024);
        in.close();
    }

    /**
     * 重命名HDFS上的文件
     * @throws Exception
     */
    @Test
    public void rename() throws Exception {
        Path oldPath = new Path("/JavaAPI/JavaAPI.txt");
        Path newPath = new Path("/JavaAPI/javaAPI.txt");
        fileSystem.rename(oldPath, newPath);
    }

    /**
     * 上传本地文件到HDFS
     * @throws Exception
     */
    @Test
    public void copyFromLocalFile() throws Exception {
        Path localPath = new Path("E://hello.txt");
        Path hdfsPath = new Path("/JavaAPI");
        fileSystem.copyFromLocalFile(localPath, hdfsPath);
    }

    /**
     * 下载HDFS上的文件
     * @throws Exception
     */
    @Test
    public void copyToLocalFile() throws Exception {
        Path localPath = new Path("E://hdfs.txt");
        Path hdfsPath = new Path("/JavaAPI/hello.txt");
        fileSystem.copyToLocalFile(hdfsPath, localPath);
    }

    /**
     * 查看某个目录下的所有文件
     * @throws Exception
     */
    @Test
    public void listFiles() throws Exception {
        FileStatus[] fileStatuses = fileSystem.listStatus(new Path("/"));
        System.out.println("文件夹/文件" + "\t" + "副本数" + "\t" + "长度" + "\t\t" + "路径");
        for(FileStatus fileStatus : fileStatuses) {
            String isDir = fileStatus.isDirectory() ? "文件夹" : "文件";
            short replication = fileStatus.getReplication();
            long len = fileStatus.getLen();
            String path = fileStatus.getPath().toString();
            System.out.println(isDir + "\t\t" + replication + "\t\t" + len + "\t\t" + path);
        }

    }

    /**
     * 删除
     * @throws Exception
     */
    @Test
    public void delete() throws Exception{
        fileSystem.delete(new Path("/JavaAPI"), true);
    }

    /**
     * 初始化
     * @throws Exception
     */
    @Before
    public void setUp() throws Exception {
        System.out.println("HDFSClient - setUp");
        configuration = new Configuration();
        fileSystem = FileSystem.get(new URI(HDFS_PATH), configuration, "hadoop-twq");
    }

    /**
     * 释放资源
     * @throws Exception
     */
    @After
    public void tearDown() throws Exception {
        configuration = null;
        fileSystem = null;
        System.out.println("HDFSApp - tearDown");
    }

}
